package com.qingcloud.df.executor.launcher.process;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.qingcloud.base.exception.WamingException;
import com.qingcloud.df.executor.constant.RuntimeConfigKeyConstant;
import com.qingcloud.df.executor.queue.ComponentMessage;
import com.qingcloud.df.executor.queue.publisher.ComponentPublisher;
import com.qingcloud.df.executor.utils.ComponentUtil;
import com.qingcloud.df.sdk.component.AbstractComponent;
import com.qingcloud.df.sdk.component.config.OutputDataItem;
import com.qingcloud.df.sdk.component.config.ResponseData;
import com.qingcloud.df.sdk.component.config.SystemRuntimeInfo;
import com.qingcloud.df.sdk.component.context.*;
import com.qingcloud.df.sdk.component.extractor.ComponentContainer;
import com.qingcloud.df.sdk.component.message.*;
import com.qingcloud.df.sdk.component.transport.handler.GatewayHandlerUtil;
import com.qingcloud.df.sdk.component.utils.ComponentProxy;
import com.qingcloud.df.sdk.component.utils.ComponentServerUtil;
import com.qingcloud.df.sdk.component.utils.SenderHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static cn.hutool.core.text.CharSequenceUtil.isNotBlank;
import static com.alibaba.fastjson.JSON.parseObject;
import static com.alibaba.fastjson.JSON.toJSONString;
import static com.qingcloud.df.executor.constant.DateFormatterConstant.NORMAL_FULL_FORMATTER;

public class ComponentExecutorHandle {

    static Logger log =  LoggerFactory.getLogger(ComponentExecutorHandle.class);
    /**
     * 数据融合核心方法
     *
     * @param componentMessage 上各组件的输出值
     */
    public static void fusion(ComponentMessage componentMessage) throws Exception {
        int instanceId = componentMessage.getInstanceId();
        ComponentValue componentValue = componentMessage.getComponentValue();
        String messageFrom = null;
        if(componentMessage.getMessageFrom()!=null) {
            messageFrom =  componentMessage.getMessageFrom();
        }
        String batchId = null;
        Integer receiveSize = 0;
        String outputContent = null;
        if(componentValue==null ){
            batchId = UUID.fastUUID().toString(true);
        } else {
            batchId = componentValue.getBatchId();
            receiveSize =  componentValue.getValues()==null ?0:componentValue.getValues().size();
        }

        CompInstRuntimeHolder runtimeHolder = null;
        try {
            //查询组件信息
            CompInstContext context = ConfigManager.loadByInstanceId(instanceId, componentMessage.getEnv());
            if(context == null) {
                log.error("组件实例id【" + instanceId + "】未找到相应配置信息");
                throw new Exception("组件实例id【" + instanceId + "】未找到相应配置信息");
            }

            SystemRuntimeInfo systemRuntimeInfo = ConfigLoader.loadRuntimeConfig(context.getInstWorkspacePath());
            if(systemRuntimeInfo!=null)
                systemRuntimeInfo.setMessageFromAddress(messageFrom);

            ComponentPublisher componentPublisher = ComponentUtil.getComponentPublisher(context);
            //记录日志和异常
            //记录执行时间和异常信息
            String startTime = LocalDateTime.now().format(NORMAL_FULL_FORMATTER);
            Exception ex = null;
            //查询组件实例对象
            AbstractComponent component = ComponentContainer.get(context.getEnv() + context.getId(), AbstractComponent.class);
            if (Objects.isNull(component)) {
                throw new WamingException("找不到组件实例ID是[" + context.getId() + "]的实例对象");
            }
            //设置组件属性
            component.setId(String.valueOf(context.getId()));
            component.setComponentId(String.valueOf(context.getComponentId()));
            component.setComponentType(context.getType());
            component.setEnv(context.getEnv());
            component.setSoftwareName(context.getSoftwareName());
            ComponentProxy componentProxy = new ComponentProxy(component, context.getInstWorkspacePath());
            runtimeHolder  =  new CompInstRuntimeHolder(context,
                    componentProxy,
                    systemRuntimeInfo);
            runtimeHolder.setRuntimeContext(runtimeHolder);
            try {
                log.debug("----------------------》正式执行组件代理对象《-----------------");
                component.beforeExecute();
                CompInstHelper.logger("----》实例编号-{}《---------", context.getId());
                CompInstHelper.logger("----》运行环境-{}《---------", componentMessage.getEnv());
                CompInstHelper.logger("----》通信地址-{}《---------", componentMessage.getMessageFrom());
                CompInstHelper.logger("----》发布主题-{}《---------", context.getTopic());
                if(componentValue !=null) {
                    CompInstHelper.logger("----》上级实例-{}《---------", componentValue.getSenderId());
                    CompInstHelper.logger("----》原始数据《---------");
                    CompInstHelper.logger(JSONUtil.toJsonPrettyStr(componentValue.getValues()));
                    CompInstHelper.logger("----》End《--------------------------------------------");
                }
                componentProxy.execute(componentValue);
                component.afterExecute();
                log.debug("----------------------》正式执行组件代理对象结束《-----------------");
            } catch (Exception e) {
                log.error("executor component error -{} - {}", instanceId,e.getMessage());
                e.printStackTrace();
                CompInstHelper.logger("组件【{}】执行出现异常, 参数值[{}]", context.getId(), JSONObject.toJSONString(componentValue));
                log.error("组件ID[{}]执行出现异常, 组件参数值是[{}]", context.getId(), JSONObject.toJSONString(componentValue));
                ex = e;
            } finally {
                Integer sendSize = 0;
                //输出
                List<Map<String, OutputDataItem>> outputData = componentProxy.allOutputData();
                if (Objects.nonNull(outputData) && !outputData.isEmpty() && outputData.size() > 0) {
                    sendSize = outputData.size();
                    //数据发送到组件队列
                    outputContent = SenderHelper.sendMultiData(context.getId(), batchId, context.getResponseTopic(),
                            outputData, componentProxy.allOutputHeader());
                }

                String endTime = LocalDateTime.now().format(NORMAL_FULL_FORMATTER);
                //组件调用记录
                ComponentRunLog runLog = new ComponentRunLog();
                runLog.setInstanceId(context.getId().toString())
                        .setBatchId(batchId)
                        .setStartTime(startTime)
                        .setCompleteTime(endTime)
                        .setResult(1)
                        .setIp(ComponentServerUtil.getLocalHost())
                        .setSendSize(sendSize)
                        .setReceiveSize(receiveSize);
                //是否有异常
                if (Objects.nonNull(ex)) { //执行错误时，需要记录错误前数据，确保后续重试完成数据处理
                    runLog.setResult(0);
                    runLog.setFailInfo(ExceptionUtil.stacktraceToString(ex));
                    runLog.setInstanceParam(ConfigLoader.loadParameterValueConfig().toJSONString());
                    runLog.setInputComponentValue(componentValue);
                }
                CompInstHelper.logger("----》【执行日志】发送成功-{}《-----------------", toJSONString(runLog));
                //是否有执行结果数据
                if (outputContent != null) {
                    ComponentValue outputValue = parseObject(outputContent, ComponentValue.class);
                    runLog.setOutputComponentValue(outputValue);
                }
                String logData = toJSONString(runLog);
                //TODO  发送给日志处理队列，重新加载处理
                //////处理完成
                ///////////////////////////
            }
            try {
                //检查返回结果
                if (outputContent !=null && componentPublisher != null && context.getTopic() != null) {
                    //发布数据到下一个节点，（所有订阅该主题的节点均可以收到该消息）
                    componentPublisher.send(context.getTopic(), messageFrom, outputContent);
                    CompInstHelper.logger("----》【结果数据】发送成功-{}《-----------------", context.getId().toString());
                }
                else {
                    CompInstHelper.logger("----》【结果数据】本次数据操作无输出信息-{}《-----------------", context.getId().toString());
                    log.debug("ID是[{}]组件实例输出结果是空的", instanceId);
                }

                //检查响应输出 响应消息来源地址
                final ResponseData responseData = componentProxy.responseData();
                if (isNotBlank(responseData.getResponseTopic()) && Objects.nonNull(responseData.getData()) && messageFrom !=null) {
                    //有响应数据，就要发送出去
                    AkkaComponentValue sendData = SenderHelper.buildSendData(context.getId(),
                            responseData.getData(), responseData.getHeaderMap());
                    sendData.setEnv(context.getEnv());
                    GatewayHandlerUtil.sendResponse(sendData,messageFrom);
                    CompInstHelper.logger("----》【响应数据】发送成功-{}《-----------------", context.getId().toString());
                }

                //状态中心
                final List<StatusCenterValue> statusCenterValues = componentProxy.statusCenterValues();
                if (Objects.nonNull(statusCenterValues) && !statusCenterValues.isEmpty()) {
                    final List<StatusCenterMessage> statusMessageList = statusCenterValues.stream().map(it -> {
                        final StatusCenterMessage message = new StatusCenterMessage();
                        BeanUtils.copyProperties(it, message);
                        message.setInstanceId(componentValue.getSenderId());
                        message.setComponentId(Integer.parseInt(componentProxy.target().getComponentId()));
                        message.setResponseTopic(componentValue.getResponseTopic());
                        message.setSendTime(LocalDateTime.now().format(NORMAL_FULL_FORMATTER));
                        message.setChannelId(componentValue.getHeaderMap().get(RuntimeConfigKeyConstant.CHANNEL_ID_KEY));
                        message.setBindIp(componentValue.getHeaderMap().get(RuntimeConfigKeyConstant.BIND_IP_KEY));
                        message.setPort(componentValue.getHeaderMap().get(RuntimeConfigKeyConstant.BIND_PORT_KEY));
                        return message;
                    }).collect(Collectors.toList());
                    //TODO
                    String statusMsg = JSONObject.toJSONString(statusMessageList);
//                    AkkaTransportUtils.sendResponse(null,messageFrom,runtimeHolder.getTransporter());
                    CompInstHelper.logger("----》【状态数据】发送成功-{}《-----------------", context.getId().toString());
                }
                log.debug("组件执行完成 ------------------ {}", context.getId().toString());
            }catch(Exception e){
                e.printStackTrace();
                log.error("数据执行成功后处理数结果失败-{}。【结果检查】【】【状态中心】",instanceId);
                CompInstHelper.logger("----》数据执行成功,但后续处理失败-{}《-----------------", context.getId().toString());
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("组件实例执行失败-{}",instanceId);
        }
    }
}
